package com.project.pulsar.base;


import com.project.pulsar.conf.PulsarConf;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 基础操作，链接、发送、手动拉取、自动监听
 */
@RestController
public class BaseController {
    @Autowired
    PulsarConf pulsarConf;

    /**
     * 生产消息
     * @param msg
     * @throws PulsarClientException
     */
    @GetMapping("/base/sendMsg")
    public MessageId sendMsg(String msg) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();

        Producer<byte[]> producer1 = pulsarFactory.newProducer()
                .topic("my-topic")

                .create();
        // 然后你就可以发送消息到指定的broker 和topic上：
        return producer1.send(msg.getBytes());
    }

    /**
     * 手动执行获取消息
     * @throws PulsarClientException
     */
    @GetMapping("/base/comsumer")
    public void comsumerByArtificial() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }

    /**
     * 自动监听消费消息
     * @throws PulsarClientException
     */
//    @Bean
//    public void comsumerByListener() throws PulsarClientException {
//        MessageListener myMessageListener = (consumer, msg) -> {
//            try {
//                System.out.println("Message received: " + new String(msg.getData()));
//                consumer.acknowledge(msg);
//            } catch (Exception e) {
//                consumer.negativeAcknowledge(msg);
//            }
//        };
//        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
//        pulsarFactory.newConsumer()
//                .topic("my-topic")
//                .subscriptionName("my-subscriptionByListener")
//                .messageListener(myMessageListener)
//                .subscribe();
//    }


}
